package server

import (
	"encoding/base64"
	"encoding/json"
	"log"
	"time"

	"github.com/corego/protoTrans/config"
	"github.com/sunface/talent"
	"github.com/valyala/fasthttp"

	"github.com/Shopify/sarama"
)

type ConsumerData struct {
	Message  string      `json:"message"`
	Expire   string      `json:"expire"`
	IOSMsg   string      `json:"iosMsg"`
	MsgID    string      `json:"msgId"`
	Recource string      `json:"resource"`
	Opid     interface{} `json:"operatorid"`
	IOSJSON  string      `json:"iosJSON"`
}

type ConsumerKafka struct {
	channels [](chan []byte)
	stop     bool
	stopC    chan bool
}

func NewConsumerKafka() *ConsumerKafka {
	consumerKafka := &ConsumerKafka{
		channels: make([](chan []byte), 0, config.Conf.KafkaConf.ChannelNum),
		stop:     false,
		stopC:    make(chan bool, 1),
	}
	return consumerKafka
}

var gConsumer *ConsumerKafka

func (consumer *ConsumerKafka) Start() {
	for i := 0; i < config.Conf.KafkaConf.ChannelNum; i++ {
		ch := make(chan []byte, config.Conf.KafkaConf.ChannelLen)
		gConsumer.channels = append(gConsumer.channels, ch)
		go func(ch chan []byte) {
			for {
				select {
				case msg, ok := <-ch:
					if ok {
						// dealData(msg)
						kafkaRoute(msg)
					}
					break
				case <-consumer.stopC:
					return
				}
			}
		}(ch)
	}
}

func (consumer *ConsumerKafka) Close() {
	consumer.stop = true
	consumer.stopC <- true
}

func initKafka() {

	conf := sarama.NewConfig()
	conf.Consumer.Retry.Backoff = 100 * time.Millisecond
	consumer, err := sarama.NewConsumer(config.Conf.KafkaConf.Addrs, conf)

	if err != nil {
		log.Printf("[FATAL] kafka host: %s\n\n\n", config.Conf.KafkaConf.Addrs)
		panic(err)
	}

	for index := 0; index < config.Conf.KafkaConf.Partition; index++ {
		partitionConsumer, err := consumer.ConsumePartition(config.Conf.KafkaConf.Topic, int32(index), sarama.OffsetNewest)
		if err != nil {
			panic(err)
		}
		go func() {
			for i := 0; ; i++ {
				if gConsumer.stop {
					break
				}
				msg := consume(partitionConsumer)
				index := i % config.Conf.KafkaConf.ChannelNum
				ch := gConsumer.channels[index]
				ch <- msg
			}
			log.Println("[INFO] Consumer is stopped")
		}()

		log.Println("[INFO] Kafka consumer started,host: ", config.Conf.KafkaConf.Addrs)
	}

}

func consume(consumer sarama.PartitionConsumer) []byte {
	msg := <-consumer.Messages()

	return msg.Value
}

func kafkaRoute(body []byte) {
	defer func() {
		if err := recover(); err != nil {
			log.Println("[WARN] route error: ", err)
			talent.PrintStack(false)
		}
	}()

	var d map[string]interface{}
	err := json.Unmarshal(body, &d)
	if err != nil {
		log.Println("[WARN] decode error: ", err, string(body))
		return
	}

	users, ok := d["operatorid"]
	if !ok {
		log.Println("[WARN] cant find push users: ", err, string(body))
		return
	}

	var uList []string
	switch u := users.(type) {
	case string:
		uList = append(uList, u)
	case []interface{}:
		for _, v := range u {
			uList = append(uList, v.(string))
		}
	}

	newRoute2mqtt(d, uList)
}

func newRoute2mqtt(d map[string]interface{}, users []string) {
	to := make(map[string]interface{})
	topic := d["resource"]
	if topic == nil {
		to["topic"] = "10001"
	}
	to["topic"] = "10001" + topic.(string)[3:]

	to["users"] = users

	var msg string
	switch md := d["message"].(type) {
	case string:
		msg = md
		break
	default:
		log.Println("[ERROR] newRoute2mqtt msg type is not string, msg is : ", md)
		return
	}

	msgData := base64.StdEncoding.EncodeToString([]byte(msg))

	to["message"] = msgData

	iosMsg := d["iosMsg"]
	if iosMsg != nil {
		to["iosMsg"] = iosMsg
	}

	iosJSON := d["iosJSON"]
	if iosJSON != nil {
		to["iosJSON"] = iosJSON
	}

	msgID := d["msgId"]
	if msgID != nil {
		to["msgID"] = msgID
	}

	expire := d["expire"]
	if expire != nil {
		to["expire"] = expire
	}

	toBuf, err := json.Marshal(to)
	if err != nil {
		log.Println("[WARN] encode to error: ", err, to)
		return
	}

	var arg fasthttp.Args
	arg.Parse(talent.Bytes2String(toBuf))
	code, body, err := client.Post(nil, config.Conf.Proto.MqttAddr, &arg)

	if code != 200 || err != nil {
		log.Println("[WARN] newRoute2mqtt route error: ", msgID, code, err, talent.Bytes2String(body), ", msg: ", msg)
		return
	}

	log.Println("[INFO] newRoute2mqtt route suc: ", msgID, code, talent.Bytes2String(body), err)
}
